package com.learning.hadoop.mapreduce.weather;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;

/**
 * ClassName: WeatherCount
 * Description: 气象数据统计
 * <p>
 * 1、计算在1949~1955年，每年温度最高的时间
 * 2、计算在1949~1955年，每年温度最高的前十天
 * 3、将每年每月中的气温排名前三的数据找出来
 * Date: 2018/3/25 20:46 【需求编号】
 *
 * @author Sam Sho
 * @version V1.0.0
 */
public class WeatherCount extends Configured implements Tool {

    public static SimpleDateFormat SDF = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    static class HotMapper extends Mapper<LongWritable, Text, KeyPari, Text> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] split = line.split("\t");
            if (split.length == 2) {
                try {
                    Date date = SDF.parse(split[0]);
                    Calendar instance = Calendar.getInstance();
                    instance.setTime(date);
                    int year = instance.get(1);
                    String hot = split[1].substring(0, split[1].indexOf("c"));
                    KeyPari keyPari = new KeyPari();
                    keyPari.setYear(year);
                    keyPari.setHot(Integer.parseInt(hot));
                    context.write(keyPari, value);
                } catch (Exception e) {
                }
            }
        }
    }

    static class HotReduce extends Reducer<KeyPari, Text, NullWritable, Text> {
        @Override
        protected void reduce(KeyPari key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            for (Text value : values) {
                context.write(NullWritable.get(), value);
            }
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        Job job = Job.getInstance(conf, WeatherCount.class.getSimpleName());
        job.setJarByClass(WeatherCount.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // map settings
        job.setMapperClass(WeatherCount.HotMapper.class);
        job.setMapOutputKeyClass(KeyPari.class);
        job.setMapOutputValueClass(Text.class);

        // reduce settings
        job.setReducerClass(WeatherCount.HotReduce.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);

        // reduce的分数与年份保持一致
        job.setNumReduceTasks(3);
        job.setPartitionerClass(FirstPartition.class);
        job.setSortComparatorClass(SortHot.class);
        job.setGroupingComparatorClass(GroupHot.class);

        boolean res = job.waitForCompletion(true);

        return res ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        if (args.length < 2) {
            return;
        }

        ToolRunner.run(new Configuration(), new WeatherCount(), args);
    }
}
